package com.sjgs.gis.fsrpc;


import com.google.protobuf.ByteString;
import com.sjgs.common.utils.DirUtils;
import com.sjgs.common.utils.KryoUtils;
import com.sjgs.gis.common.FileSystemFactory;
import com.sjgs.gis.domain.Chunk;
import com.sjgs.gis.domain.DataFile;
import com.sjgs.gis.domain.Node;
import com.sjgs.gis.domain.enumeration.DataFileStatus;
import com.sjgs.gis.fstrans.*;
import com.sjgs.gis.service.*;
import com.sjgs.gis.fs.ChunkSyncService;
import com.sjgs.gis.fs.FileSystemService;
import com.sjgs.gis.errors.InvalidChunkException;
import com.sjgs.gis.utils.RegionUtils;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.server.service.GrpcService;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.commons.util.InetUtils;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;


@GrpcService(interceptors = {HeaderServerInterceptor.class})
//, JwtServerInterceptor.class
public class FileSystemGrpcService extends FSTransServiceGrpc.FSTransServiceImplBase {

    private static final Log logger = LogFactory.getLog(FileSystemGrpcService.class.getName());

    @Value("${filesystem.grpc-chunk-size}")
    private Long grpcChunkSize;

    @Value("${filesystem.rootDir:/data}")
    private String rootDir;

    @Value("${filesystem.checkRetry:2}")
    private int checkRetry;

    @Autowired
    private FileSystemService fsService;

    @Autowired
    ChunkService chkservice;

    @Autowired
    private DataFileService dfservice;

    @Autowired
    private InetUtils inetUtils;

    @Autowired
    Node currentNode;

    @Override
    public void mkFile(com.sjgs.gis.fstrans.MakeFileRequest request,
                       io.grpc.stub.StreamObserver<com.sjgs.gis.fstrans.MakeFileResponse> responseObserver) {
        String fkey = request.getFkey();
        String dir = request.getFdir();
        Long fsize = request.getFsize();
        String storeType = request.getStoretype();
        fsService.checkFile(fkey, fsize, dir, storeType)
                .subscribe(res -> {
                    MakeFileResponse response = MakeFileResponse.newBuilder()
                            .setContext(res)
                            .build();
                    responseObserver.onNext(response);
                    responseObserver.onCompleted();
                });
    }

    @Override
    public StreamObserver<PutChunkRequest> putFile(
            StreamObserver<PutChunkResponse> responseObserver) {
        final ServerCallStreamObserver<PutChunkResponse> serverCallStreamObserver =
                (ServerCallStreamObserver<PutChunkResponse>) responseObserver;
        // @2 禁止自动流控模式，开启手动流控
        serverCallStreamObserver.disableAutoInboundFlowControl();
        // @3 背压模式流控，当消费端有足够空间时将会回调OnReadyHandler
        // 默认空间大小为65536字节
        class OnReadyHandler implements Runnable {
            private boolean wasReady = false;

            @Override
            public void run() {
                if (serverCallStreamObserver.isReady() && !wasReady) {
                    wasReady = true;
                    logger.info("READY");
                    //即发信号通知发送端发送继续发消息
                    //并拉取客户端缓冲区消息
                    serverCallStreamObserver.request(1);
                }
            }
        }
        final OnReadyHandler onReadyHandler = new OnReadyHandler();
        serverCallStreamObserver.setOnReadyHandler(onReadyHandler);
        return new StreamObserver<PutChunkRequest>() {
            @Override
            public void onNext(PutChunkRequest request) {
                writeChunk(request)
                        .doOnError(e -> {
                            responseObserver.onCompleted();
                        })
                        .subscribe(response -> {
                            try {
                                responseObserver.onNext(response);
                                if (serverCallStreamObserver.isReady()) {
                                    serverCallStreamObserver.request(1);
                                } else {
                                    onReadyHandler.wasReady = false;
                                }
                            } catch (Exception e) {
                                ;
                            }
                            logger.info("write" + request.getIndex() + request.getChunk());
                        });
            }

            @Override
            public void onError(Throwable t) {
                logger.error("put file exception");
                responseObserver.onCompleted();
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }

    @Override
    public void putChunk(PutChunkRequest request, StreamObserver<PutChunkResponse> responseObserver) {
        writeChunk(request).subscribe(response -> {
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        });
    }

    @Override
    public void fetchChunk(FetchChunkRequest request,
                           StreamObserver<FetchChunkResponse> responseObserver) {
        try {
            responseObserver.onNext(readChunk(request.getFdir(), request.getFkey(), request.getIndex()));
        } catch (IOException e) {
            logger.error(e.getMessage());
        } finally {
            responseObserver.onCompleted();
        }
    }

    @Override
    public void fetchFileInfo(com.sjgs.gis.fstrans.FileInfoRequest request,
                              io.grpc.stub.StreamObserver<com.sjgs.gis.fstrans.FileInfoResponse> responseObserver) {
        String fkey = request.getFkey();
        Long start = System.currentTimeMillis();
        dfservice.findByFkeyAndStatus(fkey, DataFileStatus.RSYNC_FINISHED.ordinal())
                .doOnError(e -> {
                    responseObserver.onCompleted();
                })
                .subscribe(df -> {
                    FileInfoResponse.Builder responseBuilder = FileInfoResponse.newBuilder();
                    String context = Base64.encodeBase64String(KryoUtils.serializeObject(df));
                    responseBuilder.setContext(context);
                    long offset = 0;
                    long index = 0;
                    long count = RegionUtils.countOf(df.getFsize(), grpcChunkSize);
                    for (index = 0; index < count; index++) {
                        offset = index * grpcChunkSize;
                        responseBuilder.addIndex(index);
                        responseBuilder.addOffset(offset);
                        if (index != count - 1) {
                            responseBuilder.addLength(Long.valueOf(grpcChunkSize));
                        } else {
                            responseBuilder.addLength(df.getFsize() - offset);
                        }
                    }
                    System.out.println("findByFkeyAndStatus cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");
                    responseObserver.onNext(responseBuilder.build());
                    responseObserver.onCompleted();
                });
    }

    @Override
    public void fetchChunkFromFile(com.sjgs.gis.fstrans.FetchChunkOfFileRequest request,
                                   io.grpc.stub.StreamObserver<com.sjgs.gis.fstrans.FetchChunkOfFileResponse> responseObserver) {
        DataFile df = KryoUtils.deserializeObject(Base64.decodeBase64(request.getContext()), DataFile.class);
        Long offset = request.getOffset();
        Long length = request.getLength();

        Mono<Resource> resourceMono = FileSystemFactory.instance().getResource(df);
        Mono<ByteBuffer> byteBufferMono = resourceMono.flatMap(resource -> {
            Flux<DataBuffer> dataBufferFlux = DataBufferUtils.read(resource, offset, new DefaultDataBufferFactory(), length.intValue());
            //Flux to Mono and return DataBuffer
            return dataBufferFlux.next().map(dataBuffer -> {
                return dataBuffer.asByteBuffer();
            });
        });
        byteBufferMono.subscribe(byteBuffer -> {
            try {
                FetchChunkOfFileResponse.Builder builder = FetchChunkOfFileResponse.newBuilder();
                builder.setFkey(df.getFkey());
                builder.setOffset(offset);
                builder.setChunk(ByteString.copyFrom(byteBuffer));
                responseObserver.onNext(builder.build());
            } catch (Exception e) {
                ;
            } finally {
                responseObserver.onCompleted();
            }
        });

    }

    /**
     * 文件检查服务
     */
    @Override
    public void checkFileCompleted(com.sjgs.gis.fstrans.CheckFileRequest request,
                                   io.grpc.stub.StreamObserver<com.sjgs.gis.fstrans.CheckFileResponse> responseObserver) {
        DataFile df = KryoUtils.deserializeObject(Base64.decodeBase64(request.getContext()), DataFile.class);

        responseObserver.onCompleted();
    }

    /**
     * 同步远程片时需要hash值
     *
     * @param chunk
     * @param response
     * @return
     */
    @Deprecated
    public FetchChunkResponse wrapChunkResponse(Chunk chunk, FetchChunkResponse response) {
        return FetchChunkResponse.newBuilder()
                .setIndex(response.getIndex())
                .setHash(chunk.getHash())
                .setChunk(response.getChunk())
                .build();
    }

    /**
     * read chunk from path
     *
     * @param fdir
     * @param fkey
     * @param index
     * @return
     * @throws IOException
     */
    public FetchChunkResponse readChunk(String fdir, String fkey, Long index) throws IOException {
        //用户空间目录
        byte[] bytes = getContent(fdir, fkey, index);
        return FetchChunkResponse.newBuilder()
                .setIndex(index)
                .setChunk(ByteString.copyFrom(bytes))
                .build();
    }

    /**
     * write chunk
     * rewrite for putChunk(DataFile datafile,Chunk chunk,FilePart filePart)@FileSystemService
     *
     * @param request
     */
    public Mono<PutChunkResponse> writeChunk(PutChunkRequest request) {
        String context = request.getContext();
        if (context.length() < 10)
            throw new InvalidChunkException();
        DataFile dataFile = KryoUtils.deserializeObject(Base64.decodeBase64(context), DataFile.class);
        String fkey = dataFile.getFkey();
        String dir = dataFile.getDir();
        Long fsize = dataFile.getFsize();
        Long index = request.getIndex();
        String chunkdir = DirUtils.getFileDir(rootDir, dir, fkey);
        Path chunkfile = Paths.get(chunkdir + "/" + index);
        //获取片字节/**/
        try {
            DirUtils.mkDir(chunkdir);
        } catch (Exception e) {
            logger.error("Can't create dir :" + chunkdir);
            throw new InvalidChunkException();
        }
        try {
            //FIXME java nio 暂时可以完成高速写入
            Files.write(chunkfile, request.getChunk().toByteArray());
        } catch (IOException e) {
            logger.error("Can't Write Chunk:" + chunkfile);
            throw new InvalidChunkException();
        }
        long count = RegionUtils.countOf(fsize, grpcChunkSize);
        Chunk chunk = new Chunk(fkey, count, index, request.getHash(), request.getStart(), request.getEnd());
        return chkservice.insertOrUpdate(chunk)
                .doOnError(e -> {
                    logger.error("insert chunk exception:" + fkey);
                    throw new InvalidChunkException();
                })
                .flatMap(chk -> {
                    return fsService.checkChunkStatus(dataFile, chk)
                            .doOnError(e -> {
                                throw new InvalidChunkException();
                            })
                            .flatMap(it -> {
                                Long chunkId = it.getId();
                                PutChunkResponse response = PutChunkResponse.newBuilder()
                                        .setChunkId(chunkId)
                                        .setChunkStatus(it.getStatus())
                                        .setChunkAddr(inetUtils.findFirstNonLoopbackAddress().getHostAddress())
                                        .build();
                                return Mono.just(response);
                            });
                });

    }

    /**
     * read chunk bytes
     *
     * @param fdir
     * @param fkey
     * @param index
     * @return
     * @throws IOException
     */
    public byte[] getContent(String fdir, String fkey, Long index) throws IOException {
        Path chunkfile = DirUtils.getChunkPath(rootDir, fdir, fkey, index);
        byte[] bytes = new byte[0];
        if (Files.exists(chunkfile)) {
            bytes = Files.readAllBytes(chunkfile);
        }
        return bytes;
    }
}
